package ls.cloud.register;

import lombok.extern.log4j.Log4j2;
import ls.cloud.conf.GRpcServerProperties;
import ls.cloud.conf.ZkProperties;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;

/**
 * Created by Administrator on 2016/12/28 0028.
 */
@Component
@Log4j2
public class ZkRpcManager implements RpcRegisterManager, InitializingBean, DisposableBean {


    @Autowired
    private GRpcServerProperties gRpcServerProperties;

    @Autowired
    private ZkProperties zkProperties;


    private ZooKeeper zk;//FIXME:断开重连

    private final static String NodeName = "GrpcServices";


    private Map<String, List<ServerInfo>> servicesCache;

    private ServerInfo myServerInfo;

    private void initMyServerInfo() throws UnknownHostException {
        String wan = gRpcServerProperties.getWan() == null ? hostAndPort() : gRpcServerProperties.getWan();
        String lan = gRpcServerProperties.getLan() == null ? hostAndPort() : gRpcServerProperties.getLan();


        ServerInfo info = new ServerInfo();
        info.setLan(lan);
        info.setWan(wan);
        this.myServerInfo = info;
    }

    //FIXME:分开内外
    private String hostAndPort() throws UnknownHostException {
        String host = null;
        if (gRpcServerProperties.getHost() != null) host = gRpcServerProperties.getHost();
        else
            host = (InetAddress.getLocalHost()).getHostName();

        return host + ":" + gRpcServerProperties.getPort();
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        initMyServerInfo();

        initZk();
    }

    private void initZk() throws IOException {
        zk = new ZooKeeper(zkProperties.getConn(), zkProperties.getSessionTimeout(), event -> {
            if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged
                    && ("/" + NodeName).equals(event.getPath()))
                reloadServices();
        }
        );

        try {
            zk.create("/" + NodeName, "".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (KeeperException.NodeExistsException ex) {
        } catch (Exception ex) {
            log.warn("initZK", ex);
        }
        reloadServices();

    }

    private synchronized void reloadServices() {//FIXME:性能优化 增量
        try {
            Map<String, List<ServerInfo>> servicesCache = new HashMap<>();
            for (String node : zk.getChildren("/" + NodeName, true)) {
                String[] nodeAndIndex = node.split("μ");
                String service = nodeAndIndex[0];
                servicesCache.computeIfAbsent(service, name -> new LinkedList<ServerInfo>()).add(
                        ServerInfo.parse(zk.getData("/" + NodeName + "/" + node, false, new Stat())));
            }
            this.servicesCache = servicesCache;

        } catch (Exception ex) {
            log.error("reloadServices", ex);
        }

    }


    @Override
    public void reg(String serviceName) {
        try {
            zk.create("/" + NodeName + "/" + serviceName + "μ", myServerInfo.toBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        } catch (Exception ex) {
            log.error("注册" + serviceName, ex);
        }
    }

    @Override
    public List<ServerInfo> serviceList(String name) {
        return servicesCache.getOrDefault(name, Collections.emptyList());
    }

    @Override
    public List<String> serviceNames() {
        return new LinkedList<>(servicesCache.keySet());
    }

    @Override
    public Map<String, List<ServerInfo>> services() {
        return servicesCache;
    }

    @Override
    public void destroy() throws Exception {

    }
}
